package com.lishem.rpc.client.runner;

import com.lishem.common.data.RpcRequest;
import com.lishem.common.data.RpcResponse;
import com.lishem.common.utils.RpcException;
import com.lishem.common.utils.SpringBeanFactory;
import com.lishem.common.utils.StatusEnum;
import com.lishem.rpc.client.cache.ServiceRouteCache;
import com.lishem.rpc.client.channel.ChannelHolder;
import com.lishem.rpc.client.channel.ProviderService;
import com.lishem.rpc.client.cluster.ClusterStrategy;
import com.lishem.rpc.client.cluster.engine.ClusterEngine;
import com.lishem.rpc.client.config.RpcClientConfiguration;
import com.lishem.rpc.client.connector.RpcClientInitializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Rpc请求管理器
 */
@Slf4j
public class RpcRequestManager {

    // 接口和 提供服务地址的缓存Map
    private static ServiceRouteCache SERVICE_ROUTE_CACHE;

    // 封装好的Netty处理链路，具体见6.7
    private static RpcClientInitializer rpcClientInitializer;
    // 负载均衡策略
    private static String CLUSTER_STRATEGY;

    private static final ConcurrentHashMap<String, ChannelHolder> channelHolderMap = new ConcurrentHashMap<>();


    public static void startRpcRequestManager(ServiceRouteCache serviceRouteCache) {
        SERVICE_ROUTE_CACHE = serviceRouteCache;
        rpcClientInitializer = SpringBeanFactory.getBean(RpcClientInitializer.class);
        CLUSTER_STRATEGY = SpringBeanFactory.getBean(RpcClientConfiguration.class).getRpcClientClusterStrategy();
    }

    /**
     * 发送客户端请求，代理对象会调用这个接口，将rpcRequest传入
     *
     * @param rpcRequest
     * @throws InterruptedException
     * @throws RpcException
     */
    public static RpcResponse sendRequest(RpcRequest rpcRequest) throws InterruptedException, RpcException {
        // 根据系统配置， 读取对应的负载均衡策略
        ClusterStrategy strategy = ClusterEngine.queryClusterStrategy(CLUSTER_STRATEGY);
        // 1. 从缓存中获取RPC服务列表信息
        List<ProviderService> providerServices = SERVICE_ROUTE_CACHE.getServiceRoutes(rpcRequest.getClassName());
        // 根据负载策略， 选取对应的实现引擎
        ProviderService targetServiceProvider = strategy.select(providerServices);
        if (targetServiceProvider != null) {
            String requestId = rpcRequest.getRequestId();
            // 3. 发起远程调用
            RpcResponse response = requestByNetty(rpcRequest, targetServiceProvider);
            log.info("Send request[{}:{}] to service provider successfully", requestId, rpcRequest.toString());
            return response;
        } else {
            throw new RpcException(StatusEnum.NOT_FOUND_SERVICE_PROVIDER);
        }
    }

    /**
     * 采用Netty进行远程调用
     */
    public static RpcResponse requestByNetty(RpcRequest rpcRequest, ProviderService providerService) {

        // 1. 创建Netty连接配置
        EventLoopGroup worker = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                .remoteAddress(providerService.getServerIp(), providerService.getNetworkPort())
                .handler(rpcClientInitializer);
        try {
            // 2. 建立连接
            ChannelFuture future = bootstrap.connect().sync();
            if (future.isSuccess()) {
                ChannelHolder channelHolder = ChannelHolder.builder()
                        .channel(future.channel())
                        .eventLoopGroup(worker)
                        .build();
                log.info("Construct a connector with service provider[{}:{}] successfully",
                        providerService.getServerIp(),
                        providerService.getNetworkPort()
                );

                // 3. 创建请求回调对象
                final RequestFuture<RpcResponse> responseFuture = new SyncRequestFuture(rpcRequest.getRequestId());
                // 4. 将请求回调放置缓存
                SyncRequestFuture.syncRequest.put(rpcRequest.getRequestId(), responseFuture);
                // 5. 根据连接通道， 下发请求信息
                ChannelFuture channelFuture = channelHolder.getChannel().writeAndFlush(rpcRequest);
                // 6. 建立回调监听
                channelFuture.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        // 7. 设置是否成功的标记
                        responseFuture.setWriteResult(future.isSuccess());
                        if (!future.isSuccess()) {
                            // 调用失败，清除连接缓存
                            SyncRequestFuture.syncRequest.remove(responseFuture.requestId());
                        }
                    }
                });
                // 8. 阻塞等待3秒
                RpcResponse result = responseFuture.get(3, TimeUnit.SECONDS);
                // 9. 移除连接缓存
                SyncRequestFuture.syncRequest.remove(rpcRequest.getRequestId());

                return result;
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
        return null;
    }
}